from pyspark.sql import SparkSession
from pyspark.sql.functions import *
from pyspark.sql.types import *

# 创建spark会话
from pyspark.sql.window import Window

spark = SparkSession \
    .builder \
    .appName("Python spark SQL") \
    .getOrCreate()


# 5.显示每年总用户数、沉默用户数（未写评论）的比例
# 读数据
user_df = spark.read. \
    json('../dataset/yelp_academic_dataset_user.sample.json')
review_df = spark.read. \
    json('../dataset/yelp_academic_dataset_review.sample.json')

# 预处理修改时间类型 YYYY-MM-DD
review_df = review_df.withColumn("date", to_timestamp(col("date"), "yyyy-MM-dd HH:mm:ss"))
review_df = review_df.withColumn("date", year(col("date")))
review_df = review_df.withColumnRenamed("user_id", "review_user_id")

print("修改格式后的df-----")
review_df.show(20)

# 并建立两个表的连接
review_user_df = review_df.join(user_df, user_df['user_id'] == review_df['review_user_id'])
print("join后的df-----")
review_user_df.show(20)

# 按照user_id和re_year分组
review_user_df = review_user_df.select('date', 'user_id', 'review_user_id')\
    .groupby('date')\
    .agg(countDistinct(col('user_id')).alias('count_year_comm'))\
    .orderBy(col('date'))
print("每年评论过的用户数-------------")
review_user_df.show()

# user表日期处理
user_df = user_df.withColumn("Join_Date",
                             to_timestamp(col("yelping_since"), "yyyy-MM-dd HH:mm:ss")) \
    .drop("yelping_since")

print("每年新加入的用户数量--------")
count_df = user_df.select('Join_Date', year('Join_Date').alias('join_year')) \
    .distinct() \
    .groupby('join_year') \
    .count() \
    .withColumn('user_count', col('count')).drop('count') \
    .orderBy(col('join_year'))
count_df.show()

print("count_df数据格式---------")
count_df.printSchema()
temp_df = spark.sql("""
select a.join_year, sum(b.user_count) as each_year_user_count
from   user_count_tb as a, user_count_tb as b
where  a.join_year>=b.join_year
group by a.join_year
""")
print("temp_df数据格式---------")
temp_df.printSchema()

print("每年用户总数----------")
temp_df.show()

# join两个表，求出比例
print('join表------')
ratio_df = count_df.join(review_user_df, temp_df['join_year'] == review_user_df['date'])
ratio_df \
    .select(col('join_year').alias('year'), (col('count_year_comm') / col('user_count')).alias('ratio')) \
    .orderBy(col('year').desc()) \
    .show()
